package com.cssl.app.dws;

import com.cssl.app.function.SplitFunction;
import com.cssl.bean.KeywordStats;
import com.cssl.utils.ClickHouseUtil;
import com.cssl.utils.CommonUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/18 22:51
 * @Desc: 关键词主题
 */
public class KeywordStatsApp {
    public static void main(String[] args) throws Exception {
        //1. 获取执行环境
        Properties properties = CommonUtils.getProperties();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2. 使用DDL方式读取数据创建表
        String pageLogSourceTopic = properties.getProperty("kafka.topic.log.dwd.page");
        String groupId = properties.getProperty("kafka.consumer.group");
        tableEnv.executeSql("create table page_view( " +
                "    `common` MAP<STRING,STRING>, " +
                "    `page` MAP<STRING,STRING>, " +
                "    `ts` BIGINT, " +
                "    `rt` as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)), " +
                "    WATERMARK FOR rt as rt - INTERVAL '1' SECOND " +
                ") WITH (" + CommonUtils.getKafkaDDL(pageLogSourceTopic, groupId) + ")");

        //3. 过滤数据 上一跳页面为"search" and 搜索词 is not null
        Table fullWordTable = tableEnv.sqlQuery("select " +
                "    page['item'] full_word, " +
                "    rt " +
                "from " +
                "    page_view " +
                "where " +
                "    page['last_page_id'] = 'search' and page['item'] is not null");

        //4. 注册UDTF,进行分词处理
        tableEnv.createTemporarySystemFunction("split_word", SplitFunction.class);
        Table wordTable = tableEnv.sqlQuery("select " +
                "    word, " +
                "    rt " +
                "from " +
                "    " + fullWordTable + ",lateral table(split_word(full_word))");

        //5. 分组,开窗,聚合
        Table resultTable = tableEnv.sqlQuery("select " +
                "    DATE_FORMAT(TUMBLE_START(rt,INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt, " +
                "    DATE_FORMAT(TUMBLE_END(rt,INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt, " +
                "    word as keyword, " +
                "    'search' as source, " +
                "    count(*) as ct, " +
                "    UNIX_TIMESTAMP()*1000 as ts " +
                "from " +
                "    " + wordTable + " " +
                "group by " +
                "    word, " +
                "    TUMBLE(rt,INTERVAL '10' SECOND)");

        //6. 将动态表转换为流
        DataStream<KeywordStats> keywordStatsDS = tableEnv.toAppendStream(resultTable, KeywordStats.class);

        //7. 将数据打印并写入clickhouse
        keywordStatsDS.print();
        keywordStatsDS.addSink(ClickHouseUtil.getSink("insert into table keyword_stats_2021 values(?,?,?,?,?,?)"));

        //8. 启动任务
        env.execute("KeywordStatsApp");
    }
}
